package yuekao9.util;

import com.alibaba.fastjson.JSON;
import org.apache.flink.api.java.tuple.*;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;
import yuekao9.entity.*;

import java.sql.*;
import java.sql.DriverManager;
import java.util.Collections;

public class AsyncIOUtil extends RichAsyncFunction<Tuple2<OrderInfo, OrderCargo>, Tuple4<OrderInfo, OrderCargo,  BaseRegionInfo, BaseDic>> {
    Connection conn = null;

    @Override
    public void open(Configuration parameters) throws Exception {
        Class.forName("org.apache.phoenix.jdbc.PhoenixDriver");
        conn = DriverManager.getConnection("jdbc:phoenix:hadoop-single:2181", "", "");
        conn.setAutoCommit(true);
    }

    @Override
    public void close() throws Exception {
        conn.close();
    }

    @Override
    public void asyncInvoke(Tuple2<OrderInfo, OrderCargo> value, ResultFuture<Tuple4<OrderInfo, OrderCargo , BaseRegionInfo, BaseDic>> resultFuture) throws Exception {
        BaseRegionInfo baseregionInfo = null;
        PreparedStatement pds = conn.prepareStatement("select * from DIM_BASE_REGION_INFO where id=?");
        pds.setObject(1, value.f0.getReceiver_city_id());
        ResultSet rs = pds.executeQuery();
        while (rs.next()) {
            String line = rs.getString("line");
//            String data = JSON.parseObject(line).getString("data");
            baseregionInfo = JSON.parseObject(line, BaseRegionInfo.class);
        }

        BaseDic basedic = null;
        PreparedStatement pds1 = conn.prepareStatement("select * from DIM_BASE_DIC where id=?");
        pds1.setObject(1, value.f1.getCargo_type());
        ResultSet rs1 = pds1.executeQuery();
        while (rs1.next()) {
            String line = rs1.getString("line");
//            String data = JSON.parseObject(line).getString("data");
            basedic = JSON.parseObject(line, BaseDic.class);
        }

        resultFuture.complete(Collections.singleton((new Tuple4<>(value.f0, value.f1, baseregionInfo, basedic))));
    }

}
